//! # MessageBus Trait 完整单元测试
//!
//! 提供 MessageBus trait 所有方法的全面测试覆盖

#[cfg(test)]
mod message_bus_tests {
    use crate::prelude::*;
    use crate::mock::{MockMessageBus, MockBehavior, test_helpers::*};
    use serde_json::json;
    use std::sync::Arc;
    use tokio::time::{timeout, Duration};
    use uuid::Uuid;

    /// 测试 MessageBus::publish 方法的所有场景
    mod publish_tests {
        use super::*;

        #[tokio::test]
        async fn test_publish_basic_message() {
            let bus = MockMessageBus::new();
            let message = create_test_message("test.topic", json!({"key": "value"}));
            
            let result = bus.publish(message.clone()).await;
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].topic(), "test.topic");
            assert_eq!(published[0].payload()["key"], "value");
        }

        #[tokio::test]
        async fn test_publish_different_message_types() {
            let bus = MockMessageBus::new();

            // 测试事件消息
            let event = create_event_message("events", json!({"type": "user_created"}));
            bus.publish(event).await.unwrap();

            // 测试命令消息
            let command = create_command_message("commands", json!({"action": "delete_user"}));
            bus.publish(command).await.unwrap();

            // 测试查询消息
            let query = create_request_message("queries", json!({"query": "get_user"}), "reply.topic");
            bus.publish(query).await.unwrap();

            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
            
            assert_eq!(published[0].metadata.attributes.message_type, MessageType::Event);
            assert_eq!(published[1].metadata.attributes.message_type, MessageType::Command);
            assert_eq!(published[2].metadata.attributes.message_type, MessageType::Query);
        }

        #[tokio::test]
        async fn test_publish_with_correlation_id() {
            let bus = MockMessageBus::new();
            let correlation_id = Uuid::new_v4();
            
            let message = create_correlated_message(
                "correlated.topic",
                json!({"data": "test"}),
                correlation_id
            );
            
            bus.publish(message).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published[0].metadata.correlation_id, Some(correlation_id));
        }

        #[tokio::test]
        async fn test_publish_different_priorities() {
            let bus = MockMessageBus::new();

            // 高优先级消息
            let high_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "high"}))
                .with_priority(MessagePriority::High)
                .build();
            bus.publish(high_priority).await.unwrap();

            // 普通优先级消息
            let normal_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "normal"}))
                .with_priority(MessagePriority::Normal)
                .build();
            bus.publish(normal_priority).await.unwrap();

            // 低优先级消息
            let low_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "low"}))
                .with_priority(MessagePriority::Low)
                .build();
            bus.publish(low_priority).await.unwrap();

            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
        }

        #[tokio::test]
        async fn test_publish_large_payload() {
            let bus = MockMessageBus::new();
            
            // 创建大负载消息
            let large_data = (0..1000).map(|i| format!("item_{}", i)).collect::<Vec<_>>();
            let message = create_test_message("large.topic", json!({"data": large_data}));
            
            let result = bus.publish(message).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_publish_empty_payload() {
            let bus = MockMessageBus::new();
            let message = create_test_message("empty.topic", json!({}));
            
            let result = bus.publish(message).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_publish_null_payload() {
            let bus = MockMessageBus::new();
            let message = create_test_message("null.topic", json!(null));
            
            let result = bus.publish(message).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_publish_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("publish", MockBehavior::new().with_failure("发布失败")).await;
            
            let message = create_test_message("fail.topic", json!({"test": "data"}));
            let result = bus.publish(message).await;
            
            assert!(result.is_err());
            assert!(result.unwrap_err().to_string().contains("发布失败"));
        }

        #[tokio::test]
        async fn test_publish_with_delay() {
            let bus = MockMessageBus::new();
            bus.set_behavior("publish", MockBehavior::new().with_delay(100)).await;
            
            let message = create_test_message("delay.topic", json!({"test": "data"}));
            let start = std::time::Instant::now();
            
            bus.publish(message).await.unwrap();
            
            assert!(start.elapsed().as_millis() >= 100);
        }

        #[tokio::test]
        async fn test_publish_after_close() {
            let bus = MockMessageBus::new();
            bus.close().await.unwrap();
            
            let message = create_test_message("closed.topic", json!({"test": "data"}));
            let result = bus.publish(message).await;
            
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::subscribe 方法的所有场景
    mod subscribe_tests {
        use super::*;

        #[tokio::test]
        async fn test_subscribe_basic_topic() {
            let bus = MockMessageBus::new();
            let result = bus.subscribe("test.topic").await;
            
            assert!(result.is_ok());
            let subscriber = result.unwrap();
            assert_eq!(subscriber.pattern(), "test.topic");
            assert!(subscriber.is_active());
        }

        #[tokio::test]
        async fn test_subscribe_wildcard_patterns() {
            let bus = MockMessageBus::new();

            // 单级通配符
            let sub1 = bus.subscribe("events.*").await.unwrap();
            assert_eq!(sub1.pattern(), "events.*");

            // 多级通配符
            let sub2 = bus.subscribe("events.**").await.unwrap();
            assert_eq!(sub2.pattern(), "events.**");

            // 全匹配通配符
            let sub3 = bus.subscribe("*").await.unwrap();
            assert_eq!(sub3.pattern(), "*");
        }

        #[tokio::test]
        async fn test_subscribe_complex_patterns() {
            let bus = MockMessageBus::new();

            let patterns = vec![
                "user.events.login.*",
                "system.*.critical.**",
                "app.module.*.events",
                "**",
            ];

            for pattern in patterns {
                let result = bus.subscribe(pattern).await;
                assert!(result.is_ok());
            }
        }

        #[tokio::test]
        async fn test_subscribe_same_pattern_multiple_times() {
            let bus = MockMessageBus::new();
            
            let sub1 = bus.subscribe("same.topic").await.unwrap();
            let sub2 = bus.subscribe("same.topic").await.unwrap();
            
            // 两个订阅者应该是不同的实例
            assert_ne!(sub1.subscription_id(), sub2.subscription_id());
        }

        #[tokio::test]
        async fn test_subscribe_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("subscribe", MockBehavior::new().with_failure("订阅失败")).await;
            
            let result = bus.subscribe("fail.topic").await;
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_subscribe_with_delay() {
            let bus = MockMessageBus::new();
            bus.set_behavior("subscribe", MockBehavior::new().with_delay(50)).await;
            
            let start = std::time::Instant::now();
            let result = bus.subscribe("delay.topic").await;
            
            assert!(result.is_ok());
            assert!(start.elapsed().as_millis() >= 50);
        }

        #[tokio::test]
        async fn test_subscribe_after_close() {
            let bus = MockMessageBus::new();
            bus.close().await.unwrap();
            
            let result = bus.subscribe("closed.topic").await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::unsubscribe 方法的所有场景
    mod unsubscribe_tests {
        use super::*;

        #[tokio::test]
        async fn test_unsubscribe_existing_subscription() {
            let bus = MockMessageBus::new();
            let subscriber = bus.subscribe("test.topic").await.unwrap();
            let subscription_id = subscriber.subscription_id();
            
            let result = bus.unsubscribe(subscription_id).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_unsubscribe_non_existing_subscription() {
            let bus = MockMessageBus::new();
            let result = bus.unsubscribe("non-existing-id").await;
            assert!(result.is_ok()); // 应该静默成功
        }

        #[tokio::test]
        async fn test_unsubscribe_empty_id() {
            let bus = MockMessageBus::new();
            let result = bus.unsubscribe("").await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_unsubscribe_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("unsubscribe", MockBehavior::new().with_failure("取消订阅失败")).await;
            
            let result = bus.unsubscribe("test-id").await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::subscriber_count 方法的所有场景
    mod subscriber_count_tests {
        use super::*;

        #[tokio::test]
        async fn test_subscriber_count_no_subscribers() {
            let bus = MockMessageBus::new();
            let count = bus.subscriber_count("empty.topic").await.unwrap();
            assert_eq!(count, 0);
        }

        #[tokio::test]
        async fn test_subscriber_count_with_subscribers() {
            let bus = MockMessageBus::new();
            
            // 为同一主题创建多个订阅者
            let _sub1 = bus.subscribe("counted.topic").await.unwrap();
            let _sub2 = bus.subscribe("counted.topic").await.unwrap();
            let _sub3 = bus.subscribe("other.topic").await.unwrap();
            
            let count = bus.subscriber_count("counted.topic").await.unwrap();
            assert_eq!(count, 2);
            
            let other_count = bus.subscriber_count("other.topic").await.unwrap();
            assert_eq!(other_count, 1);
        }

        #[tokio::test]
        async fn test_subscriber_count_after_unsubscribe() {
            let bus = MockMessageBus::new();
            
            let sub1 = bus.subscribe("test.topic").await.unwrap();
            let sub2 = bus.subscribe("test.topic").await.unwrap();
            
            let count_before = bus.subscriber_count("test.topic").await.unwrap();
            assert_eq!(count_before, 2);
            
            bus.unsubscribe(sub1.subscription_id()).await.unwrap();
            
            let count_after = bus.subscriber_count("test.topic").await.unwrap();
            assert_eq!(count_after, 1);
        }

        #[tokio::test]
        async fn test_subscriber_count_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("subscriber_count", MockBehavior::new().with_failure("计数失败")).await;
            
            let result = bus.subscriber_count("fail.topic").await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::list_topics 方法的所有场景
    mod list_topics_tests {
        use super::*;

        #[tokio::test]
        async fn test_list_topics_empty() {
            let bus = MockMessageBus::new();
            let topics = bus.list_topics().await.unwrap();
            assert_eq!(topics.len(), 0);
        }

        #[tokio::test]
        async fn test_list_topics_with_published_messages() {
            let bus = MockMessageBus::new();
            
            // 发布到不同主题的消息
            let message1 = create_test_message("topic1", json!({"test": 1}));
            let message2 = create_test_message("topic2", json!({"test": 2}));
            let message3 = create_test_message("topic1", json!({"test": 3})); // 重复主题
            
            bus.publish(message1).await.unwrap();
            bus.publish(message2).await.unwrap();
            bus.publish(message3).await.unwrap();
            
            let topics = bus.list_topics().await.unwrap();
            assert_eq!(topics.len(), 2);
            assert!(topics.contains(&"topic1".to_string()));
            assert!(topics.contains(&"topic2".to_string()));
        }

        #[tokio::test]
        async fn test_list_topics_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("list_topics", MockBehavior::new().with_failure("列表获取失败")).await;
            
            let result = bus.list_topics().await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::has_subscribers 方法的所有场景
    mod has_subscribers_tests {
        use super::*;

        #[tokio::test]
        async fn test_has_subscribers_false() {
            let bus = MockMessageBus::new();
            let has_subs = bus.has_subscribers("empty.topic").await.unwrap();
            assert!(!has_subs);
        }

        #[tokio::test]
        async fn test_has_subscribers_true() {
            let bus = MockMessageBus::new();
            let _subscriber = bus.subscribe("active.topic").await.unwrap();
            
            let has_subs = bus.has_subscribers("active.topic").await.unwrap();
            assert!(has_subs);
        }

        #[tokio::test]
        async fn test_has_subscribers_after_unsubscribe() {
            let bus = MockMessageBus::new();
            let subscriber = bus.subscribe("temp.topic").await.unwrap();
            
            let has_before = bus.has_subscribers("temp.topic").await.unwrap();
            assert!(has_before);
            
            bus.unsubscribe(subscriber.subscription_id()).await.unwrap();
            
            let has_after = bus.has_subscribers("temp.topic").await.unwrap();
            assert!(!has_after);
        }

        #[tokio::test]
        async fn test_has_subscribers_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("has_subscribers", MockBehavior::new().with_failure("检查失败")).await;
            
            let result = bus.has_subscribers("fail.topic").await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::close 方法的所有场景
    mod close_tests {
        use super::*;

        #[tokio::test]
        async fn test_close_clean_shutdown() {
            let bus = MockMessageBus::new();
            let _subscriber = bus.subscribe("test.topic").await.unwrap();
            
            let result = bus.close().await;
            assert!(result.is_ok());
            assert!(bus.is_closed().await);
        }

        #[tokio::test]
        async fn test_close_multiple_times() {
            let bus = MockMessageBus::new();
            
            let result1 = bus.close().await;
            let result2 = bus.close().await;
            
            assert!(result1.is_ok());
            assert!(result2.is_ok()); // 多次关闭应该成功
        }

        #[tokio::test]
        async fn test_close_with_active_operations() {
            let bus = MockMessageBus::new();
            let subscriber = bus.subscribe("active.topic").await.unwrap();
            
            // 模拟正在进行的操作
            let message = create_test_message("active.topic", json!({"data": "test"}));
            bus.publish(message).await.unwrap();
            
            let result = bus.close().await;
            assert!(result.is_ok());
            
            // 订阅者应该被关闭
            assert!(!subscriber.is_active());
        }

        #[tokio::test]
        async fn test_close_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("close", MockBehavior::new().with_failure("关闭失败")).await;
            
            let result = bus.close().await;
            assert!(result.is_err());
        }
    }

    /// 测试 MessageBus::stats 方法的所有场景
    mod stats_tests {
        use super::*;

        #[tokio::test]
        async fn test_stats_empty_bus() {
            let bus = MockMessageBus::new();
            let stats = bus.stats().await.unwrap();
            
            assert_eq!(stats.total_published, 0);
            assert_eq!(stats.total_received, 0);
            assert_eq!(stats.total_failed, 0);
            assert_eq!(stats.active_subscribers, 0);
            assert_eq!(stats.active_topics, 0);
        }

        #[tokio::test]
        async fn test_stats_with_activity() {
            let bus = MockMessageBus::new();
            
            // 创建订阅者
            let _sub1 = bus.subscribe("topic1").await.unwrap();
            let _sub2 = bus.subscribe("topic2").await.unwrap();
            
            // 发布消息
            let message1 = create_test_message("topic1", json!({"test": 1}));
            let message2 = create_test_message("topic2", json!({"test": 2}));
            bus.publish(message1).await.unwrap();
            bus.publish(message2).await.unwrap();
            
            let stats = bus.stats().await.unwrap();
            assert_eq!(stats.total_published, 2);
            assert_eq!(stats.active_subscribers, 2);
            assert_eq!(stats.active_topics, 2);
        }

        #[tokio::test]
        async fn test_stats_failure() {
            let bus = MockMessageBus::new();
            bus.set_behavior("stats", MockBehavior::new().with_failure("统计失败")).await;
            
            let result = bus.stats().await;
            assert!(result.is_err());
        }
    }

    /// 集成测试：测试方法之间的交互
    mod integration_tests {
        use super::*;

        #[tokio::test]
        async fn test_publish_subscribe_flow() {
            let bus = MockMessageBus::new();
            
            // 创建订阅者
            let subscriber = bus.subscribe("integration.test").await.unwrap();
            
            // 发布消息
            let message = create_test_message(
                "integration.test",
                json!({"flow": "publish_subscribe"})
            );
            bus.publish(message.clone()).await.unwrap();
            
            // 验证消息被正确路由
            let published = bus.get_published_messages_for_topic("integration.test").await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].payload()["flow"], "publish_subscribe");
            
            // 验证统计信息
            let stats = bus.stats().await.unwrap();
            assert_eq!(stats.total_published, 1);
            assert_eq!(stats.active_subscribers, 1);
        }

        #[tokio::test]
        async fn test_lifecycle_management() {
            let bus = MockMessageBus::new();
            
            // 阶段1：启动和订阅
            let sub1 = bus.subscribe("lifecycle.test").await.unwrap();
            let sub2 = bus.subscribe("lifecycle.test").await.unwrap();
            assert_eq!(bus.subscriber_count("lifecycle.test").await.unwrap(), 2);
            
            // 阶段2：消息发布
            let message = create_test_message("lifecycle.test", json!({"stage": "active"}));
            bus.publish(message).await.unwrap();
            
            // 阶段3：部分取消订阅
            bus.unsubscribe(sub1.subscription_id()).await.unwrap();
            assert_eq!(bus.subscriber_count("lifecycle.test").await.unwrap(), 1);
            
            // 阶段4：完全关闭
            bus.close().await.unwrap();
            assert!(bus.is_closed().await);
            assert!(!sub2.is_active());
        }

        #[tokio::test]
        async fn test_error_recovery() {
            let bus = MockMessageBus::new();
            
            // 设置发布失败
            bus.set_behavior("publish", MockBehavior::new().with_failure("临时失败")).await;
            
            let message = create_test_message("error.test", json!({"test": "recovery"}));
            let result = bus.publish(message.clone()).await;
            assert!(result.is_err());
            
            // 清除失败行为
            bus.set_behavior("publish", MockBehavior::new()).await;
            
            // 重试应该成功
            let retry_result = bus.publish(message).await;
            assert!(retry_result.is_ok());
        }
    }

    /// 并发测试
    mod concurrent_tests {
        use super::*;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[tokio::test]
        async fn test_concurrent_publish() {
            let bus = Arc::new(MockMessageBus::new());
            let mut handles = vec![];
            
            for i in 0..10 {
                let bus_clone = bus.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "concurrent.publish",
                        json!({"id": i})
                    );
                    bus_clone.publish(message).await
                });
                handles.push(handle);
            }
            
            // 等待所有任务完成
            for handle in handles {
                assert!(handle.await.unwrap().is_ok());
            }
            
            // 验证所有消息都被发布
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 10);
        }

        #[tokio::test]
        async fn test_concurrent_subscribe() {
            let bus = Arc::new(MockMessageBus::new());
            let mut handles = vec![];
            
            for _i in 0..5 {
                let bus_clone = bus.clone();
                let handle = tokio::spawn(async move {
                    bus_clone.subscribe("concurrent.subscribe").await
                });
                handles.push(handle);
            }
            
            // 等待所有订阅完成
            for handle in handles {
                assert!(handle.await.unwrap().is_ok());
            }
            
            // 验证订阅者数量
            let count = bus.subscriber_count("concurrent.subscribe").await.unwrap();
            assert_eq!(count, 5);
        }

        #[tokio::test]
        async fn test_concurrent_mixed_operations() {
            let bus = Arc::new(MockMessageBus::new());
            let publish_count = Arc::new(AtomicUsize::new(0));
            let subscribe_count = Arc::new(AtomicUsize::new(0));
            
            let mut handles = vec![];
            
            // 并发发布
            for i in 0..5 {
                let bus_clone = bus.clone();
                let counter = publish_count.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "mixed.operations",
                        json!({"operation": "publish", "id": i})
                    );
                    if bus_clone.publish(message).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            // 并发订阅
            for _i in 0..3 {
                let bus_clone = bus.clone();
                let counter = subscribe_count.clone();
                let handle = tokio::spawn(async move {
                    if bus_clone.subscribe("mixed.operations").await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            // 等待所有操作完成
            for handle in handles {
                handle.await.unwrap();
            }
            
            assert_eq!(publish_count.load(Ordering::Relaxed), 5);
            assert_eq!(subscribe_count.load(Ordering::Relaxed), 3);
        }
    }
}